package com.my.job;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by Scarlett on 2018/8/14 0014.
 */
public class Job {
    private Source source;
    private Sink sink;
    private List<Step> steps = new ArrayList<>();

    public <S> Job source(Source<S> source) {
        this.source = source;
        return this;
    }

    public <S, T> Job step(Step<S, T> step) {
        this.steps.add(step);
        return this;
    }

    public <S> Job sink(Sink<S> sink) {
        this.sink = sink;
        return this;
    }

    public void execute() {
        Object read = this.source.read();

        Object source = read;
        Object target = read;
        for (Step step : steps) {
            target = step.process(source);
            source = target;
        }

        this.sink.write(target);
    }
}
